package com.hs.kfk.redis;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.Set;

/**
 * @Description: 消费者基于redis改造
 * @Author 胡尚
 * @Date: 2024/8/8 11:19
 */
public class RedisConsumer {

    private final static String BOOTSTRAP_SERVER = "worker1:9092,worker2:9092,worker3:9092";
    private final static String TOPIC_NAME = "disTopic";
    private final static String CONSUMER_GROUP = "test";

    public static void main(String[] args) {
        // 设置消费者相关的属性
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 创建消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) {
            // 消费者拉取消息
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(100));

            // 处理消息
            for (ConsumerRecord<String, String> record : consumerRecords) {
                int partition = record.partition();
                long offset = record.offset();
                String topic = record.topic();
                String redisKey = "redisKeyPrefex" + topic + partition + CONSUMER_GROUP;
                String key = record.key();
                String message = record.value();
                // TODO 通过redisKey 查询redis中的offset，如果 redisOffset > offset 就表示重复消费了

                // TODO 消费消息

                // TODO 存入redis中  redisKey + offset
            }

            consumer.commitSync();
        }
    }
}
